package com.lmq

//import com.lmq.filterLen.spark
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

/**
 * TODO: NOT USED.
 */
object combiner {


  Logger.getLogger("org.apache.spark")
    .setLevel(Level.WARN)

  val spark = SparkSession.builder()
    .master("local[*]")
    .appName("Test")
    .getOrCreate()


  def main(args: Array[String]): Unit = {
    val utils = new utils()

    val schema = StructType(
      Array(
        StructField("SessionId", IntegerType, nullable = true),
        StructField("TimeStr", StringType, nullable = true),
        StructField("ItemId", IntegerType, nullable = true),
        StructField("Context", StringType, nullable = true)
      )
    )
    val df: DataFrame = spark.read
      .schema(schema)
      .option("header",value = true)
      .csv("D:\\pythonProject\\pythonProject\\ComparisonWithYoochoose\\src\\yoochoose-clicks.csv")
//      .csv("file:///home/iptv/yoochoose/yoochoose-clicks.dat")
    df.show(false)

    val timeTransfer = udf((x:String)=>  utils.timeStr2Tsp(x) )
    df.select(
      col("SessionId"),
      col("TimeStr"),
      col("ItemId"),
      col("Context"),
      timeTransfer( col("TimeStr") ).alias("Time")
    )
      .createOrReplaceTempView("allData")

    val v: DataFrame = spark.sql(""" select * ,rank() over (partition by SessionId order by Time) as
    `TimeRank` from allData""")
    v.createTempView("V")
    spark.sql(
      """select v.* from
        |(select ItemId from
        |( select ItemId, count(1) as cnt
        |from v
        |group by ItemId) A where cnt>=5) B
        |left join v
        |on B.ItemId=v.ItemId
        |""".stripMargin)
    //        .count())
    //        .show(1000,truncate = false)

    //      val xtable =
    spark.sql(
      """select w.SessionId,w.ItemId,w.Context,w.Time,
        |rank() over (partition by SessionId order by Time) as TimeRank
        |from
        |(select v.* from
        |(select ItemId from
        |( select ItemId, count(1) as cnt
        |from v
        |group by ItemId) A where cnt>=5) B
        |left join v
        |on B.ItemId=v.ItemId) w
        |""".stripMargin).createTempView("xtable")
    //        .show(false)
    spark.sql(
      """
        |select xtable.SessionId,xtable.ItemId,xtable.Context,xtable.Time
        |from
        |(select SessionId from
        |(select SessionId,sum(1) as cnts from xtable group by SessionId) groupedtable where cnts>1
        |) ltable
        |left join xtable
        |on ltable.SessionId=xtable.SessionId
        |""".stripMargin)
      .show(1000,false)



    // remove sessions which the length is less than 1 and apearance is less than 5.

    //    v.coalesce(1).write
    //      .option("header","true")
    //      .csv("file:///home/iptv/yoochoose/Reslt.csv")

    //      .toPandas()\
    //    .to_csv("helloResult.csv",index=None)







  }

}
